package com.bblei.hbaseDemo;

import org.apache.commons.collections.CollectionUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;

public class HbaseDao {
    public static final byte[] DEFAUT_FAMIY = Bytes.toBytes("1");
    public static final byte[] MY_FAMIY = Bytes.toBytes("2");
    private static Configuration conf = HBaseConfiguration.create();


    static {
        conf.set("hbase.rootdir", "hdfs://10.30.182.2:8020/hbase");
        // 设置Zookeeper,直接设置IP地址
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        conf.set("hbase.zookeeper.quorum", "10.30.182.2,10.30.74.11,10.30.190.13,10.30.184.22");
        conf.setInt("hbase.client.pause", 20);
        conf.setInt("hbase.client.retries", 11);
        conf.setBoolean("hbase.ipc.client.tcpnodelay", true);
        conf.setInt("ipc.ping.interval.", 4000);

    }

    // 创建表
    public static void createTable(String tablename) throws Exception {
        Connection connection = ConnectionFactory.createConnection(conf);
        Admin   admin = connection.getAdmin();

        TableName tableNameObj = TableName.valueOf(tablename);

        if (admin.tableExists(tableNameObj)) {
            System.out.println("Table exists!");
            System.exit(0);
        } else {
            HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf(tablename));
            HColumnDescriptor columnDescriptor = new HColumnDescriptor(
                    DEFAUT_FAMIY);
            columnDescriptor.setMaxVersions(1);
            tableDescriptor.addFamily(columnDescriptor);
            //列族2
            HColumnDescriptor mycolumnDescriptor = new HColumnDescriptor(
                    MY_FAMIY);
            mycolumnDescriptor.setMaxVersions(1);
            tableDescriptor.addFamily(mycolumnDescriptor);
            admin.createTable(tableDescriptor);
            System.out.println("create table success!");
        }
        admin.close();
        connection.close();
    }

    // 删除表
    public static void deleteTable(String tableName) {
        try {
            Connection connection = ConnectionFactory.createConnection(conf);
            Admin admin = connection.getAdmin();
            TableName table = TableName.valueOf(tableName);
            admin.disableTable(table);
            admin.deleteTable(table);
            System.out.println("delete table " + tableName + " ok.");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    // 插入记录
    public static void addRecord(Map<String, LinkedList<Put>> alltableput) throws IOException {


        Connection connection = ConnectionFactory.createConnection(conf);
        Table table = null;
        try {
            if (null != alltableput && !alltableput.isEmpty()) {
                for (String tbn : alltableput.keySet()) {//tbn key
                    LinkedList<Put> links = alltableput.get(tbn);
                    if (CollectionUtils.isEmpty(links)) {
                        System.out.println("HBaseFlushThread  error  links is null,tablename is " + tbn + ", links is" + links);
                        continue;
                    }
                    //hbase需要的表名称
                    String tablename = tbn;
                    table = connection.getTable(TableName.valueOf(tablename));
                    //hbase需要的数据
                    // 表名   rowkey
                    table.put(links);
                    System.out.println("insert success!");
                }
                table.close();
            }

        } catch (Exception e) {
            System.out.println( e);

        } finally {
            if (table != null) {
                try {
                    table.close();
                } catch (IOException e) {
                    System.out.println(e);
                }
            }
        }
    }

    public static void queryData() throws Exception {

        Connection connection = ConnectionFactory.createConnection(conf);
        Table table = connection.getTable(TableName.valueOf("person"));

        Get get = new Get(Bytes.toBytes("9223370523159175807460041765206018")); //rowkey

//        get.addColumn(Bytes.toBytes("1"),Bytes.toBytes("COUNT"));//只查询一列数据
        Result result = table.get(get); //得到所有列的数据
        //族  列
        System.out.println(Bytes.toString(result.getValue(Bytes.toBytes("1"), Bytes.toBytes("COUNT"))));
        System.out.println(Bytes.toString(result.getValue(Bytes.toBytes("1"), Bytes.toBytes("HEIGHT"))));
        System.out.println(Bytes.toString(result.getValue(Bytes.toBytes("1"), Bytes.toBytes("WIDTH"))));
    }


    public static void scanData() throws Exception {

        Connection connection = ConnectionFactory.createConnection(conf);
        Table table = connection.getTable(TableName.valueOf("person"));

        //全表扫描 如果不是设置指定列 会加载整个表
        Scan scan = new Scan();
        //scan.addFamily(Bytes.toBytes("info")); // 只扫描指定的族
        //scan.addColumn(Bytes.toBytes("info"), Bytes.toBytes("password"));// 只扫描指定的行
//        scan.setStartRow(Bytes.toBytes("wangsf_0"));//设置rowkey 的开始到结束的时间，rowkey 的字典序列 区间
//        scan.setStopRow(Bytes.toBytes("wangwu"));
        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            //族  列
            System.out.println( Bytes.toString( result.getRow()));//得到rowkey
            System.out.println(Bytes.toString(result.getValue(Bytes.toBytes("1"), Bytes.toBytes("COUNT"))));
            System.out.println(Bytes.toString(result.getValue(Bytes.toBytes("1"), Bytes.toBytes("HEIGHT"))));
            System.out.println(Bytes.toString(result.getValue(Bytes.toBytes("1"), Bytes.toBytes("WIDTH"))));
        }
    }


    /**
     * 全表扫描的过滤器
     * 列值过滤器
     *
     * @throws Exception
     */

    public static void scanDataByFilter1() throws Exception {

        Connection connection = ConnectionFactory.createConnection(conf);
        Table table = connection.getTable(TableName.valueOf("person"));

        // 创建全表扫描的scan
        Scan scan = new Scan();


        //过滤器：列值过滤器 族 列  比较规则  标准值
        SingleColumnValueFilter filter = new SingleColumnValueFilter(Bytes.toBytes("1"),
                Bytes.toBytes("HEIGHT"), CompareFilter.CompareOp.NOT_EQUAL,
                Bytes.toBytes("0"));

        // 设置过滤器
        scan.setFilter(filter);

        // 打印结果集
        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            //族  列
            System.out.println( Bytes.toString( result.getRow()));//得到rowkey
            System.out.println(Bytes.toString(result.getValue(Bytes.toBytes("1"), Bytes.toBytes("COUNT"))));
            System.out.println(Bytes.toString(result.getValue(Bytes.toBytes("1"), Bytes.toBytes("HEIGHT"))));
            System.out.println(Bytes.toString(result.getValue(Bytes.toBytes("1"), Bytes.toBytes("WIDTH"))));
        }

    }

    /**
     * 匹配列名前缀
     * @throws Exception
     */

    public static void scanDataByFilter2() throws Exception {

        Connection connection = ConnectionFactory.createConnection(conf);
        Table table = connection.getTable(TableName.valueOf("person"));

        // 创建全表扫描的scan
        Scan scan = new Scan();

        //获取 列名为 COUNT 的所有数据
        ColumnPrefixFilter filter = new ColumnPrefixFilter(Bytes.toBytes("COUNT"));

       // Filter filter1
        // 设置过滤器
        scan.setFilter(filter);

        // 打印结果集
        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            //族  列
            System.out.println( Bytes.toString( result.getRow()));//得到rowkey
            System.out.println(Bytes.toString(result.getValue(Bytes.toBytes("1"), Bytes.toBytes("COUNT"))));
            System.out.println(Bytes.toString(result.getValue(Bytes.toBytes("1"), Bytes.toBytes("HEIGHT"))));
            System.out.println(Bytes.toString(result.getValue(Bytes.toBytes("1"), Bytes.toBytes("WIDTH"))));
        }

    }


    /**
     * rowkey过滤器
     * @throws Exception
     */

    public static void scanDataByFilter3() throws Exception {


        Connection connection = ConnectionFactory.createConnection(conf);
        Table table = connection.getTable(TableName.valueOf("person"));

        // 创建全表扫描的scan
        Scan scan = new Scan();
        //922337052315917开头的rowkey
      //  RowFilter filter = new RowFilter(CompareFilter.CompareOp.EQUAL, new RegexStringComparator("^922337052315917"));

        //提取rowkey以18结尾数据
//        RowFilter filter = new RowFilter(CompareFilter.CompareOp.EQUAL,new RegexStringComparator(".*18$"));

        //提取rowkey以包含201407的数据
        RowFilter filter = new RowFilter(CompareFilter.CompareOp.EQUAL,new SubstringComparator("9223370523159175807460041765206018"));
        //

        // 设置过滤器
        scan.setFilter(filter);
        // 打印结果集
        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            //族  列
            System.out.println( Bytes.toString( result.getRow()));//得到rowkey
            System.out.println(Bytes.toString(result.getValue(Bytes.toBytes("1"), Bytes.toBytes("COUNT"))));
            System.out.println(Bytes.toString(result.getValue(Bytes.toBytes("1"), Bytes.toBytes("HEIGHT"))));
            System.out.println(Bytes.toString(result.getValue(Bytes.toBytes("1"), Bytes.toBytes("WIDTH"))));
        }


    }

    /**
     * 过滤器集合
     * @throws Exception
     */
    public void scanDataByFilter4() throws Exception {

        Connection connection = ConnectionFactory.createConnection(conf);
        Table table = connection.getTable(TableName.valueOf("person"));

        // 创建全表扫描的scan
        Scan scan = new Scan();
        //过滤器集合：MUST_PASS_ALL（and）,MUST_PASS_ONE(or)
        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ONE);
        //匹配rowkey以wangsenfeng开头的
        RowFilter filter = new RowFilter(CompareFilter.CompareOp.EQUAL, new RegexStringComparator("^wangsenfeng"));
        //匹配name的值等于wangsenfeng
        SingleColumnValueFilter filter2 = new SingleColumnValueFilter(Bytes.toBytes("info"),
                Bytes.toBytes("name"), CompareFilter.CompareOp.EQUAL,
                Bytes.toBytes("zhangsan"));
        filterList.addFilter(filter);
        filterList.addFilter(filter2);
        // 设置过滤器
        scan.setFilter(filterList);
        // 打印结果集
        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            System.out.println("rowkey：" + Bytes.toString(result.getRow()));
            System.out.println("info:name："
                    + Bytes.toString(result.getValue(Bytes.toBytes("info"),
                    Bytes.toBytes("name"))));
            // 判断取出来的值是否为空
            if (result.getValue(Bytes.toBytes("info"), Bytes.toBytes("age")) != null) {
                System.out.println("info:age："
                        + Bytes.toInt(result.getValue(Bytes.toBytes("info"),
                        Bytes.toBytes("age"))));
            }
            // 判断取出来的值是否为空
            if (result.getValue(Bytes.toBytes("info"), Bytes.toBytes("sex")) != null) {
                System.out.println("infi:sex："
                        + Bytes.toInt(result.getValue(Bytes.toBytes("info"),
                        Bytes.toBytes("sex"))));
            }
            // 判断取出来的值是否为空
            if (result.getValue(Bytes.toBytes("info2"), Bytes.toBytes("name")) != null) {
                System.out
                        .println("info2:name："
                                + Bytes.toString(result.getValue(
                                Bytes.toBytes("info2"),
                                Bytes.toBytes("name"))));
            }
            // 判断取出来的值是否为空
            if (result.getValue(Bytes.toBytes("info2"), Bytes.toBytes("age")) != null) {
                System.out.println("info2:age："
                        + Bytes.toInt(result.getValue(Bytes.toBytes("info2"),
                        Bytes.toBytes("age"))));
            }
            // 判断取出来的值是否为空
            if (result.getValue(Bytes.toBytes("info2"), Bytes.toBytes("sex")) != null) {
                System.out.println("info2:sex："
                        + Bytes.toInt(result.getValue(Bytes.toBytes("info2"),
                        Bytes.toBytes("sex"))));
            }
        }

    }

    public static void main(String[] args) throws Exception {


//            HbaseDao.createTable("person");
//
//        LinkedList<Put> puts = new LinkedList<Put>();
//        Map<String, LinkedList<Put>> datas = new HashMap<String, LinkedList<Put>>();
//        String rowKey = (Long.MAX_VALUE - Long.valueOf("1513695600000")) + "460041765206018";
//        Put put = new Put(Bytes.toBytes(rowKey));
//        String count_value = null;
//
//        for (int i=1;i<101;i++) {
//            count_value = "bbleiii" + String.valueOf(i);
//            put.addColumn(DEFAUT_FAMIY, Bytes.toBytes("COUNT"), Bytes.toBytes(count_value));
//            System.out.println(count_value);
//            puts.add(put);
//        }
//        datas.put("person",puts);
//        HbaseDao.addRecord(datas);

//        HbaseDao.queryData();

//        HbaseDao.scanData();
        HbaseDao.scanDataByFilter3();
//      HbaseDao.addRecord("testTb", "001", "info", "name", "zhangsan");
//      HbaseDao.addRecord("testTb", "001", "info", "age", "20");
//      HbaseDao.deleteTable("person");
    }

}
